package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013,4, Constant.TOPIC_DWD_TRADE_CART_ADD);
    }


    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {

        // 1. 读取 topic_db 数据
        readOdsDb(tEnv,groupId);

        // 2. 过滤出加购数据
        Table table = cart_info(tEnv);

        //3. 在kafka的链接表
        getTableResult(tEnv);

        // . 写出到 kafka
        table.insertInto(Constant.TOPIC_DWD_TRADE_CART_ADD).execute();

    }

    private static TableResult getTableResult(StreamTableEnvironment tEnv) {
        return tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_CART_ADD + "\n" +
                " (\n" +
                "id string,\n" +
                "user_id string,\n" +
                "sku_id string,\n" +
                "cart_price string,\n" +
                "sku_num bigint,\n" +
                "sku_name string,\n" +
                "is_checked string,\n" +
                "create_time string,\n" +
                "operate_time string,\n" +
                "is_ordered string,\n" +
                "order_time string,\n" +
                "source_type string ,\n" +
                "source_id string,\n" +
                "ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_CART_ADD));
    }

    private static Table cart_info(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['cart_price'] cart_price,\n" +
                "if(`type`='insert',cast(`data`['sku_num'] as bigint),cast(`data`['sku_num'] as bigint)-cast(`old`['sku_num'] as bigint)) as sku_num,\n" +
                "`data`['sku_name'] sku_name,\n" +
                "`data`['is_checked'] is_checked,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['is_ordered'] is_ordered,\n" +
                "`data`['order_time'] order_time," +
                "`data`['source_type'] source_type," +
                "`data`['source_id'] source_id," +
                "ts \n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='cart_info'\n" +
                "and `type`='insert' or(`type`='update' and `old`['sku_num'] is not null)\n" +
                "and cast(`data`['sku_num'] as bigint) > cast (`old`['sku_num'] as bigint)");
    }


}
